package org.lds.message.center.alipay;

import org.apache.commons.lang3.StringUtils;
import org.lds.message.center.alipay.constant.AlipayConstant;
import org.lds.message.center.alipay.enums.AlipayErrorEnum;
import org.lds.message.center.alipay.properties.AlipayProperties;
import org.lds.message.center.common.AbstractMessageService;
import org.lds.message.center.common.MessageHandler;
import org.lds.message.center.common.model.request.AlipayRequest;
import org.lds.message.center.common.util.AssertUtil;
import org.lds.message.center.plugin.core.kafka.KafkaProducer;

import java.util.Objects;

public class AlipayMessageService extends AbstractMessageService<AlipayRequest> {
    private final AlipayProperties alipayProperties;
    private final KafkaProducer kafkaProducer;

    public AlipayMessageService(MessageHandler<AlipayRequest> messageHandler, AlipayProperties alipayProperties, KafkaProducer kafkaProducer) {
        super(messageHandler);
        this.alipayProperties = alipayProperties;
        this.kafkaProducer = kafkaProducer;
    }

    @Override
    public String channelName() {
        return AlipayConstant.CHANNEL_NAME;
    }

    @Override
    public String channelType() {
        return AlipayConstant.CHANNEL_TYPE;
    }

    @Override
    public Class<AlipayRequest> supportClass() {
        return AlipayRequest.class;
    }

    @Override
    public void check(AlipayRequest request) {
        AssertUtil.assertTrue(alipayProperties.isEnabled(), AlipayErrorEnum.PLUGIN_CLOSED);
        AssertUtil.assertTrue(Objects.nonNull(request), AlipayErrorEnum.REQUEST_INVALID);
    }

    @Override
    protected void store(AlipayRequest request) {
        kafkaProducer.send(alipayProperties.getTopic(), request);
    }

}
